package boot.spring.service.impl;

import boot.spring.model.po.TrialClient;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static boot.spring.constant.TrailContant.prefix;

public class RedisTrailSubscriptionThread extends Thread {
    private static final Logger logger = LoggerFactory.getLogger(RedisTrailSubscriptionThread.class);

    private TrialClient trialClient;
    private RedisConnectionFactory connectionFactory;
    private RedisTemplate<String, Object> redisTemplate;

    private RedisMessageListenerContainer container = new RedisMessageListenerContainer();

    public RedisTrailSubscriptionThread(TrialClient trialClient, RedisConnectionFactory connectionFactory, RedisTemplate<String, Object> redisTemplate) {
        this.trialClient = trialClient;
        this.connectionFactory = connectionFactory;
        this.redisTemplate = redisTemplate;
    }

    public void run() {
        List<String> channels = trialClient.getIntersectionIdList().stream()
                .map(intersectionId -> prefix + "trail_updates:" + intersectionId)
                .collect(Collectors.toList());
        container.setConnectionFactory(connectionFactory);
        List<PatternTopic> topics = channels.stream().map(PatternTopic::new).collect(Collectors.toList());
        container.addMessageListener(
                (message, pattern) -> {
                    try {
                        String channel = new String(pattern);
                        //message是车辆ID
                        //String key = prefix + "trail:" + channel.split(":")[2] + ":" + message;
                        String key = prefix + "trail:" + channel.split(":")[2];
                        boolean exist = redisTemplate.hasKey(key);
                        if (!exist) {
                            return;
                        }
                        List<Object> trails = redisTemplate.opsForList().range(key, 0, -1);
                        Map<String, Object> trailMap = new HashMap<>();
                        trailMap.put("result", trails);
                        String trailJson = JSON.toJSONString(trailMap);
                        trialClient.getTrailWebSocketServer().sendMessage(trailJson);
                    } catch (Exception e) {
                        logger.warn("给{}发送消息失败。{}", trialClient.getSessionId(), e.getMessage(), e);
                    }
                },
                topics
        );
        container.afterPropertiesSet();
        container.start();
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            logger.warn("Thread interrupted, exiting...");
        }
    }

    public void stopThread() {
        container.stop();
        this.interrupt();
    }
}
